package br.edu.ufcg.ccc.wordcount;

import java.io.IOException;
import java.util.List;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.util.StringUtils;

import br.edu.ufcg.ccc.util.TwitterParseUtil;

/**
 * 
 * @author victorclf
 * @author pedroyossis
 */
@SuppressWarnings("deprecation")
public class TrioCountMap extends MapReduceBase implements
		Mapper<LongWritable, Text, Text, IntWritable> {
	private final static IntWritable one = new IntWritable(1);
	private Text word = new Text();

	public void map(LongWritable key, Text value,
			OutputCollector<Text, IntWritable> output, Reporter reporter)
			throws IOException {
		String line = value.toString();
		line = TwitterParseUtil.preformatLine(line);
										
		List<String> rawTokens = TwitterParseUtil.tokenize(line);
		List<String> tokens = TwitterParseUtil.getValidTokens(rawTokens);
		
		for (int i = 0; i < tokens.size(); ++i) {
			for (int j = 0; j < tokens.size(); ++j) {
				for (int k = 0; k < tokens.size(); ++k) {
					if (i != j && i != k && j != k) {
						word.set(StringUtils.join(" ", new String[]{tokens.get(i), tokens.get(j), tokens.get(k)}));
						output.collect(word, one);
					}
				}
			}
		}
	}
}